"""test_watershed.py - tests the watershed function"""

import math
import unittest

import numpy as np
import pytest
from scipy import ndimage as ndi

import skimage.measure
from skimage._shared.filters import gaussian
from skimage.feature import peak_local_max
from skimage.measure import label

from skimage.segmentation._watershed import watershed

eps = 1e-12
# fmt: off
blob = np.array([[255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255],
                 [255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255],
                 [255, 255, 255, 255, 255, 204, 204, 204, 204, 204, 204, 255, 255, 255, 255, 255],
                 [255, 255, 255, 204, 204, 183, 153, 153, 153, 153, 183, 204, 204, 255, 255, 255],
                 [255, 255, 204, 183, 153, 141, 111, 103, 103, 111, 141, 153, 183, 204, 255, 255],
                 [255, 255, 204, 153, 111,  94,  72,  52,  52,  72,  94, 111, 153, 204, 255, 255],
                 [255, 255, 204, 153, 111,  72,  39,   1,   1,  39,  72, 111, 153, 204, 255, 255],
                 [255, 255, 204, 183, 141, 111,  72,  39,  39,  72, 111, 141, 183, 204, 255, 255],
                 [255, 255, 255, 204, 183, 141, 111,  72,  72, 111, 141, 183, 204, 255, 255, 255],
                 [255, 255, 255, 255, 204, 183, 141,  94,  94, 141, 183, 204, 255, 255, 255, 255],
                 [255, 255, 255, 255, 255, 204, 153, 103, 103, 153, 204, 255, 255, 255, 255, 255],
                 [255, 255, 255, 255, 204, 183, 141,  94,  94, 141, 183, 204, 255, 255, 255, 255],
                 [255, 255, 255, 204, 183, 141, 111,  72,  72, 111, 141, 183, 204, 255, 255, 255],
                 [255, 255, 204, 183, 141, 111,  72,  39,  39,  72, 111, 141, 183, 204, 255, 255],
                 [255, 255, 204, 153, 111,  72,  39,   1,   1,  39,  72, 111, 153, 204, 255, 255],
                 [255, 255, 204, 153, 111,  94,  72,  52,  52,  72,  94, 111, 153, 204, 255, 255],
                 [255, 255, 204, 183, 153, 141, 111, 103, 103, 111, 141, 153, 183, 204, 255, 255],
                 [255, 255, 255, 204, 204, 183, 153, 153, 153, 153, 183, 204, 204, 255, 255, 255],
                 [255, 255, 255, 255, 255, 204, 204, 204, 204, 204, 204, 255, 255, 255, 255, 255],
                 [255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255],
                 [255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255]])
# fmt: on


def diff(a, b):
    if not isinstance(a, np.ndarray):
        a = np.asarray(a)
    if not isinstance(b, np.ndarray):
        b = np.asarray(b)
    if (0 in a.shape) and (0 in b.shape):
        return 0.0
    b[a == 0] = 0
    if a.dtype in [np.complex64, np.complex128] or b.dtype in [
        np.complex64,
        np.complex128,
    ]:
        a = np.asarray(a, np.complex128)
        b = np.asarray(b, np.complex128)
        t = ((a.real - b.real) ** 2).sum() + ((a.imag - b.imag) ** 2).sum()
    else:
        a = np.asarray(a)
        a = a.astype(np.float64)
        b = np.asarray(b)
        b = b.astype(np.float64)
        t = ((a - b) ** 2).sum()
    return math.sqrt(t)


class TestWatershed(unittest.TestCase):
    eight = np.ones((3, 3), bool)

    def test_watershed01(self):
        "watershed 1"
        data = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.uint8,
        )
        markers = np.array(
            [
                [-1, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 1, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.int8,
        )
        out = watershed(data, markers, self.eight)
        expected = np.array(
            [
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
            ]
        )
        error = diff(expected, out)
        assert error < eps

    def test_watershed02(self):
        "watershed 2"
        data = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.uint8,
        )
        markers = np.array(
            [
                [-1, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 1, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.int8,
        )
        out = watershed(data, markers)
        error = diff(
            [
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, 1, 1, 1, -1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, -1, 1, 1, 1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
            ],
            out,
        )
        self.assertTrue(error < eps)

    def test_watershed03(self):
        "watershed 3"
        data = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.uint8,
        )
        markers = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 2, 0, 3, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, -1],
            ],
            np.int8,
        )
        out = watershed(data, markers)
        error = diff(
            [
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, 0, 2, 0, 3, 0, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, 0, 2, 0, 3, 0, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
            ],
            out,
        )
        self.assertTrue(error < eps)

    def test_watershed04(self):
        "watershed 4"
        data = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.uint8,
        )
        markers = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 2, 0, 3, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, -1],
            ],
            np.int8,
        )
        out = watershed(data, markers, self.eight)
        error = diff(
            [
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, 2, 2, 0, 3, 3, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
            ],
            out,
        )
        self.assertTrue(error < eps)

    def test_watershed05(self):
        "watershed 5"
        data = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 0, 1, 0, 1, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.uint8,
        )
        markers = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 3, 0, 2, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, -1],
            ],
            np.int8,
        )
        out = watershed(data, markers, self.eight)
        error = diff(
            [
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, 3, 3, 0, 2, 2, -1],
                [-1, 3, 3, 0, 2, 2, -1],
                [-1, 3, 3, 0, 2, 2, -1],
                [-1, 3, 3, 0, 2, 2, -1],
                [-1, 3, 3, 0, 2, 2, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
            ],
            out,
        )
        self.assertTrue(error < eps)

    def test_watershed06(self):
        "watershed 6"
        data = np.array(
            [
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 0, 0, 0, 1, 0],
                [0, 1, 1, 1, 1, 1, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
            ],
            np.uint8,
        )
        markers = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 1, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0],
                [-1, 0, 0, 0, 0, 0, 0],
            ],
            np.int8,
        )
        out = watershed(data, markers, self.eight)
        error = diff(
            [
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, 1, 1, 1, 1, 1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
                [-1, -1, -1, -1, -1, -1, -1],
            ],
            out,
        )
        self.assertTrue(error < eps)

    def test_watershed07(self):
        "A regression test of a competitive case that failed"
        data = blob
        mask = data != 255
        markers = np.zeros(data.shape, int)
        markers[6, 7] = 1
        markers[14, 7] = 2
        out = watershed(data, markers, self.eight, mask=mask)
        #
        # The two objects should be the same size, except possibly for the
        # border region
        #
        size1 = np.sum(out == 1)
        size2 = np.sum(out == 2)
        self.assertTrue(abs(size1 - size2) <= 6)

    def test_watershed08(self):
        "The border pixels + an edge are all the same value"
        data = blob.copy()
        data[10, 7:9] = 141
        mask = data != 255
        markers = np.zeros(data.shape, int)
        markers[6, 7] = 1
        markers[14, 7] = 2
        out = watershed(data, markers, self.eight, mask=mask)
        #
        # The two objects should be the same size, except possibly for the
        # border region
        #
        size1 = np.sum(out == 1)
        size2 = np.sum(out == 2)
        self.assertTrue(abs(size1 - size2) <= 6)

    def test_watershed09(self):
        """Test on an image of reasonable size

        This is here both for timing (does it take forever?) and to
        ensure that the memory constraints are reasonable
        """
        image = np.zeros((1000, 1000))
        coords = np.random.uniform(0, 1000, (100, 2)).astype(int)
        markers = np.zeros((1000, 1000), int)
        idx = 1
        for x, y in coords:
            image[x, y] = 1
            markers[x, y] = idx
            idx += 1

        image = gaussian(image, sigma=4, mode='reflect')
        watershed(image, markers, self.eight)
        ndi.watershed_ift(image.astype(np.uint16), markers, self.eight)

    def test_watershed10(self):
        "watershed 10"
        data = np.array(
            [[1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1, 1]], np.uint8
        )
        markers = np.array(
            [[1, 0, 0, 2], [0, 0, 0, 0], [0, 0, 0, 0], [3, 0, 0, 4]], np.int8
        )
        out = watershed(data, markers, self.eight)
        error = diff([[1, 1, 2, 2], [1, 1, 2, 2], [3, 3, 4, 4], [3, 3, 4, 4]], out)
        self.assertTrue(error < eps)

    def test_watershed11(self):
        '''Make sure that all points on this plateau are assigned to closest seed'''
        # https://github.com/scikit-image/scikit-image/issues/803
        #
        # Make sure that no point in a level image is farther away
        # from its seed than any other
        #
        image = np.zeros((21, 21))
        markers = np.zeros((21, 21), int)
        markers[5, 5] = 1
        markers[5, 10] = 2
        markers[10, 5] = 3
        markers[10, 10] = 4

        structure = np.array(
            [[False, True, False], [True, True, True], [False, True, False]]
        )
        out = watershed(image, markers, structure)
        i, j = np.mgrid[0:21, 0:21]
        d = np.dstack(
            [
                np.sqrt((i.astype(float) - i0) ** 2, (j.astype(float) - j0) ** 2)
                for i0, j0 in ((5, 5), (5, 10), (10, 5), (10, 10))
            ]
        )
        dmin = np.min(d, 2)
        self.assertTrue(np.all(d[i, j, out[i, j] - 1] == dmin))

    def test_watershed12(self):
        "The watershed line"
        data = np.array(
            [
                [
                    203,
                    255,
                    203,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                ],
                [
                    203,
                    255,
                    203,
                    153,
                    153,
                    153,
                    102,
                    102,
                    102,
                    102,
                    102,
                    102,
                    153,
                    153,
                    153,
                    153,
                ],
                [
                    203,
                    255,
                    203,
                    203,
                    153,
                    153,
                    102,
                    102,
                    77,
                    0,
                    102,
                    102,
                    153,
                    153,
                    203,
                    203,
                ],
                [
                    203,
                    255,
                    255,
                    203,
                    153,
                    153,
                    153,
                    102,
                    102,
                    102,
                    102,
                    153,
                    153,
                    203,
                    203,
                    255,
                ],
                [
                    203,
                    203,
                    255,
                    203,
                    203,
                    203,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    203,
                    203,
                    255,
                    255,
                ],
                [
                    153,
                    203,
                    255,
                    255,
                    255,
                    203,
                    203,
                    203,
                    203,
                    203,
                    203,
                    203,
                    203,
                    255,
                    255,
                    203,
                ],
                [
                    153,
                    203,
                    203,
                    203,
                    255,
                    255,
                    255,
                    255,
                    255,
                    255,
                    255,
                    255,
                    255,
                    255,
                    203,
                    203,
                ],
                [
                    153,
                    153,
                    153,
                    203,
                    203,
                    203,
                    203,
                    203,
                    255,
                    203,
                    203,
                    203,
                    203,
                    203,
                    203,
                    153,
                ],
                [
                    102,
                    102,
                    153,
                    153,
                    153,
                    153,
                    203,
                    203,
                    255,
                    203,
                    203,
                    255,
                    203,
                    153,
                    153,
                    153,
                ],
                [
                    102,
                    102,
                    102,
                    102,
                    102,
                    153,
                    203,
                    255,
                    255,
                    203,
                    203,
                    203,
                    203,
                    153,
                    102,
                    153,
                ],
                [
                    102,
                    51,
                    51,
                    102,
                    102,
                    153,
                    203,
                    255,
                    203,
                    203,
                    153,
                    153,
                    153,
                    153,
                    102,
                    153,
                ],
                [
                    77,
                    51,
                    51,
                    102,
                    153,
                    153,
                    203,
                    255,
                    203,
                    203,
                    203,
                    153,
                    102,
                    102,
                    102,
                    153,
                ],
                [
                    77,
                    0,
                    51,
                    102,
                    153,
                    203,
                    203,
                    255,
                    203,
                    255,
                    203,
                    153,
                    102,
                    51,
                    102,
                    153,
                ],
                [
                    77,
                    0,
                    51,
                    102,
                    153,
                    203,
                    255,
                    255,
                    203,
                    203,
                    203,
                    153,
                    102,
                    0,
                    102,
                    153,
                ],
                [
                    102,
                    0,
                    51,
                    102,
                    153,
                    203,
                    255,
                    203,
                    203,
                    153,
                    153,
                    153,
                    102,
                    102,
                    102,
                    153,
                ],
                [
                    102,
                    102,
                    102,
                    102,
                    153,
                    203,
                    255,
                    203,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                    153,
                ],
            ]
        )
        markerbin = data == 0
        marker = label(markerbin)
        ws = watershed(data, marker, connectivity=2, watershed_line=True)
        for lab, area in zip(range(4), [34, 74, 74, 74]):
            self.assertTrue(np.sum(ws == lab) == area)

    def test_watershed_input_not_modified(self):
        """Test to ensure input markers are not modified."""
        image = np.random.default_rng().random(size=(21, 21))
        markers = np.zeros((21, 21), dtype=np.uint8)
        markers[[5, 5, 15, 15], [5, 15, 5, 15]] = [1, 2, 3, 4]
        original_markers = np.copy(markers)
        result = watershed(image, markers)
        np.testing.assert_equal(original_markers, markers)
        assert not np.all(result == markers)


def test_compact_watershed():
    # in this test, when compactness is greater than zero the watershed line
    # is labeled with the closest marker (label=2)
    # when compactness is zero the watershed line is labeled with
    # the marker that reaches it first (label=1)
    # because it has a zero cost path to the line.
    image = np.zeros((5, 6))
    image[:, 3] = 2  # watershed line
    image[:, 4:] = 1
    seeds = np.zeros((5, 6), dtype=int)
    seeds[2, 0] = 1
    seeds[2, 5] = 2
    compact = watershed(image, seeds, compactness=0.01)
    expected = np.array(
        [
            [1, 1, 1, 2, 2, 2],
            [1, 1, 1, 2, 2, 2],
            [1, 1, 1, 2, 2, 2],
            [1, 1, 1, 2, 2, 2],
            [1, 1, 1, 2, 2, 2],
        ],
        dtype=int,
    )
    np.testing.assert_equal(compact, expected)
    normal = watershed(image, seeds)
    expected = np.array(
        [
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
        ],
        dtype=int,
    )
    np.testing.assert_equal(normal, expected)

    # checks that compact watershed labels with watershed lines are
    # a subset of the labels from compact watershed for this specific example
    compact_wsl = watershed(image, seeds, compactness=0.01, watershed_line=True)
    difference = compact_wsl != compact
    difference[compact_wsl == 0] = False

    assert not np.any(difference)


def test_watershed_with_markers_offset():
    """
    Check edge case behavior reported in gh-6632

    While we initially viewed the behavior described in gh-6632 [1]_ as a bug,
    we have reverted that decision in gh-7661. See [2]_ for an explanation.
    So this test now actually asserts the behavior reported in gh-6632 as
    correct.

    .. [1] https://github.com/scikit-image/scikit-image/issues/6632.
    .. [2] https://github.com/scikit-image/scikit-image/issues/7661#issuecomment-2645810807
    """
    # Generate an initial image with two overlapping circles
    x, y = np.indices((80, 80))
    x1, y1, x2, y2 = 28, 28, 44, 52
    r1, r2 = 16, 20
    mask_circle1 = (x - x1) ** 2 + (y - y1) ** 2 < r1**2
    mask_circle2 = (x - x2) ** 2 + (y - y2) ** 2 < r2**2
    image = np.logical_or(mask_circle1, mask_circle2)

    # Now we want to separate the two objects in image
    # Generate the markers as local maxima of the distance to the background
    # and then apply an y-offset
    distance = ndi.distance_transform_edt(image)
    coords = peak_local_max(distance, footprint=np.ones((3, 3)), labels=image)
    coords[:, 0] += 6
    mask = np.zeros(distance.shape, dtype=bool)
    mask[tuple(coords.T)] = True
    markers, _ = ndi.label(mask)

    labels = watershed(-distance, markers, mask=image)

    props = skimage.measure.regionprops(labels, intensity_image=-distance)

    # Generally, assert that the smaller object could only conquer a thin line
    # in the direction of the positive gradient
    assert props[0].extent == 1
    expected_region = np.arange(start=-10, stop=0, dtype=float).reshape(-1, 1)
    np.testing.assert_equal(props[0].image_intensity, expected_region)

    # Assert pixel count from reviewed reproducing example in bug report
    assert props[0].num_pixels == 10
    assert props[1].num_pixels == 1928


def test_watershed_simple_basin_overspill():
    """
    Test edge case behavior when markers spill over into another basin / compete.

    While we initially viewed the behavior described in gh-6632 [1]_ as a bug,
    we have reverted that decision in gh-7661. See [2]_ for an explanation.
    So this test now actually asserts the behavior reported in gh-6632 as
    correct.

    .. [1] https://github.com/scikit-image/scikit-image/issues/6632.
    .. [2] https://github.com/scikit-image/scikit-image/issues/7661#issuecomment-2645810807
    """
    # Scenario 1
    # fmt: off
    image =    np.array([[6, 5, 4, 3, 0, 3, 0, 1, 2],
                         [6, 5, 4, 3, 0, 3, 0, 1, 2]])
    markers =  np.array([[0, 1, 0, 0, 0, 0, 0, 2, 0],
                         [0, 0, 0, 0, 0, 0, 0, 0, 0]])
    expected = np.array([[1, 1, 2, 2, 2, 2, 2, 2, 2],
                         [2, 2, 2, 2, 2, 2, 2, 2, 2]])
    # fmt: on
    result = watershed(image, markers=markers)
    np.testing.assert_equal(result, expected)

    # Scenario 2
    image = -np.array([1, 2, 2, 2, 2, 2, 3])
    markers = np.array([1, 0, 0, 0, 0, 0, 2])
    expected = np.array([1, 2, 2, 2, 2, 2, 2])
    result = watershed(image, markers=markers, mask=image != 0)
    np.testing.assert_array_equal(result, expected)


def test_watershed_evenly_distributed_overspill():
    """
    Edge case: Basins should be distributed evenly between contesting markers.

    Markers should be prevented from spilling over into another basin and
    conquering it against other markers with the same claim, just because they
    get to the basin one step earlier.
    """
    # Scenario 1: markers start with the same value
    image =    np.array([0, 2, 1, 1, 1, 1, 1, 1, 2, 0])  # fmt: skip
    markers =  np.array([1, 0, 0, 0, 0, 0, 0, 0, 0, 2])  # fmt: skip
    expected = np.array([1, 1, 1, 1, 1, 2, 2, 2, 2, 2])  # fmt: skip
    result = watershed(image, markers=markers)
    np.testing.assert_equal(result, expected)

    # Scenario 2: markers start with the different values
    image =    np.array([2, 2, 1, 1, 1, 1, 1, 1, 2, 0])  # fmt: skip
    expected = np.array([1, 1, 1, 1, 1, 2, 2, 2, 2, 2])  # fmt: skip
    result = watershed(image, markers=markers)
    np.testing.assert_equal(result, expected)


def test_markers_on_maxima():
    """Check that markers placed at maxima don't conquer other pixels.

    Regression test for gh-7661 [1]_.

    .. [1] https://github.com/scikit-image/scikit-image/issues/7661
    """
    image =    np.array([[0, 1, 2, 3, 4, 5, 4],
                         [0, 1, 2, 3, 4, 4, 4]])  # fmt: skip
    markers =  np.array([[1, 0, 0, 0, 0, 2, 0],
                         [0, 0, 0, 0, 0, 0, 0]])  # fmt: skip
    expected = np.array([[1, 1, 1, 1, 1, 2, 1],
                         [1, 1, 1, 1, 1, 1, 1]])  # fmt: skip
    result = watershed(image, markers=markers)
    np.testing.assert_equal(result, expected)


def test_numeric_seed_watershed():
    """Test that passing just the number of seeds to watershed works."""
    image = np.zeros((5, 6))
    image[:, 3:] = 1
    compact = watershed(image, 2, compactness=0.01)
    expected = np.array(
        [
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
            [1, 1, 1, 1, 2, 2],
        ],
        dtype=np.int32,
    )
    np.testing.assert_equal(compact, expected)


@pytest.mark.parametrize(
    'dtype',
    [np.uint8, np.int8, np.uint16, np.int16, np.uint32, np.int32, np.uint64, np.int64],
)
def test_watershed_output_dtype(dtype):
    image = np.zeros((100, 100))
    markers = np.zeros((100, 100), dtype)
    out = watershed(image, markers)
    assert out.dtype == markers.dtype


def test_incorrect_markers_shape():
    image = np.ones((5, 6))
    markers = np.ones((5, 7))
    with pytest.raises(ValueError):
        watershed(image, markers)


def test_incorrect_mask_shape():
    image = np.ones((5, 6))
    mask = np.ones((5, 7))
    with pytest.raises(ValueError):
        watershed(image, markers=4, mask=mask)


def test_markers_in_mask():
    data = blob
    mask = data != 255
    out = watershed(data, 25, connectivity=2, mask=mask)
    # There should be no markers where the mask is false
    assert np.all(out[~mask] == 0)


def test_no_markers():
    data = blob
    mask = data != 255
    out = watershed(data, mask=mask)
    assert np.max(out) == 2


def test_connectivity():
    """
    Watershed segmentation should output different result for
    different connectivity
    when markers are calculated where None is supplied.
    Issue = 5084
    """
    # Generate a dummy BrightnessTemperature image
    x, y = np.indices((406, 270))
    x1, y1, x2, y2, x3, y3, x4, y4 = 200, 208, 300, 120, 100, 100, 340, 208
    r1, r2, r3, r4 = 100, 50, 40, 80
    mask_circle1 = (x - x1) ** 2 + (y - y1) ** 2 < r1**2
    mask_circle2 = (x - x2) ** 2 + (y - y2) ** 2 < r2**2
    mask_circle3 = (x - x3) ** 2 + (y - y3) ** 2 < r3**2
    mask_circle4 = (x - x4) ** 2 + (y - y4) ** 2 < r4**2
    image = np.logical_or(mask_circle1, mask_circle2)
    image = np.logical_or(image, mask_circle3)
    image = np.logical_or(image, mask_circle4)

    # calculate distance in discrete increase
    DummyBT = ndi.distance_transform_edt(image)
    DummyBT_dis = np.around(DummyBT / 12, decimals=0) * 12
    # calculate the mask
    Img_mask = np.where(DummyBT_dis == 0, 0, 1)

    # segments for connectivity 1 and 2
    labels_c1 = watershed(
        200 - DummyBT_dis, mask=Img_mask, connectivity=1, compactness=0.01
    )
    labels_c2 = watershed(
        200 - DummyBT_dis, mask=Img_mask, connectivity=2, compactness=0.01
    )

    # assertions
    assert np.unique(labels_c1).shape[0] == 6
    assert np.unique(labels_c2).shape[0] == 5

    # checking via area of each individual segment.
    for lab, area in zip(range(6), [61824, 3653, 20467, 11097, 1301, 11278]):
        assert np.sum(labels_c1 == lab) == area

    for lab, area in zip(range(5), [61824, 3653, 20466, 12386, 11291]):
        assert np.sum(labels_c2 == lab) == area
